# consumer.py
from kafka import KafkaConsumer
import os
import json

# 配置 Kafka SASL/PLAIN 认证
os.environ['KAFKA_OPTS'] = '-Djava.security.auth.login.config=kafka_client_jaas.conf'

# Kafka 消费者配置
consumer = KafkaConsumer(
    'offence',                                # 订阅的 Kafka 主题
    bootstrap_servers=['localhost:9093'],   # Kafka 服务器地址和端口
    security_protocol='SASL_PLAINTEXT',     # 安全协议
    sasl_mechanism='PLAIN',                 # SASL 机制
    sasl_plain_username='user',             # Kafka 用户名
    sasl_plain_password='user-secret',      # Kafka 密码
    auto_offset_reset='earliest',           # 消费者读取消息的起始位置
    enable_auto_commit=True,                # 自动提交消费偏移量
    group_id='testGroup',                    # 消费者组 ID
)

print("等待从 Kafka 主题 'test' 接收消息...")

# 接收消息
for message in consumer:
    try:
        data = json.loads(message.value.decode('utf-8'))
        # 处理数据
    except json.JSONDecodeError as e:
        print(f"JSON解码错误：{e}")
    print(f"消息偏移量: {message.offset}, 消息内容: {data}, resId: {data['resId']}")
    

consumer.close()
